package com.mysite.producer;

import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

@Service
public final class ProducerService {

	int count = 0;
	Logger logger = LogManager.getLogger(this.getClass());

	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;

	private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

	public void produce() {
		String time = format.format(new Date());
		for (int i = 1; i < 5; i++) {
			count++;
			String message = count + " message produced at : " + time;
			logger.info(message);
			ListenableFuture<SendResult<String, String>> future = kafkaTemplate.sendDefault(message);
			future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
				@Override
				public void onSuccess(SendResult<String, String> arg0) {
					System.err.println(arg0);
				}

				@Override
				public void onFailure(Throwable arg0) {
					System.err.println(arg0);
				}
			});
		}
	}
}
